package service.impl;

import client.DatabaseClient;
import dto.Request;
import model.NodeInfo;
import service.ReplicationService;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

/**
 * 数据复制服务实现类
 */
@Slf4j
public class ReplicationServiceImpl implements ReplicationService {

    private final Map<String, DatabaseClient> slaveClients = new ConcurrentHashMap<>();
    private ExecutorService replicationThreadPool;

    @Override
    public void initialize(List<NodeInfo> slaveNodes) {
        log.info("初始化数据复制服务，从节点列表: {}", slaveNodes);
        if (slaveNodes == null || slaveNodes.isEmpty()) {
            log.info("无从节点，跳过复制线程池初始化");
            this.replicationThreadPool = null;
            return;
        }
        this.replicationThreadPool = Executors.newFixedThreadPool(slaveNodes.size());
        
        for (NodeInfo slave : slaveNodes) {
            DatabaseClient client = new DatabaseClient(slave.getHost(), slave.getPort());
            if (client.connect()) {
                slaveClients.put(slave.getNodeId(), client);
                log.info("成功连接到从节点: {}", slave.getNodeId());
            } else {
                log.warn("连接从节点失败: {}", slave.getNodeId());
            }
        }
    }

    @Override
    public void replicate(Request request) {
        if (slaveClients.isEmpty()) {
            return;
        }
        // 标记为复制请求，防止从节点再次复制
        request.setReplicated(true);

        for (Map.Entry<String, DatabaseClient> entry : slaveClients.entrySet()) {
            String nodeId = entry.getKey();
            DatabaseClient client = entry.getValue();
            if (client.isConnected()) {
                log.debug("向从节点 {} 复制请求: {}", nodeId, request.getType());
                client.sendRequestAsync(request); // 假设客户端有异步发送方法
            } else {
                log.warn("从节点 {} 已断开连接，无法复制", nodeId);
                // 这里可以加入重连逻辑
            }
        }
    }

    @Override
    public void replicateAsync(Request request) {
        if (replicationThreadPool == null) {
            // 没有从节点，无需同步
            return;
        }
        replicationThreadPool.submit(() -> replicate(request));
    }

    @Override
    public void shutdown() {
        log.info("正在关闭数据复制服务...");
        replicationThreadPool.shutdown();
        slaveClients.values().forEach(DatabaseClient::disconnect);
        log.info("数据复制服务已关闭");
    }

    @Override
    public Map<String, Boolean> getSlaveStatus() {
        return slaveClients.entrySet().stream()
                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().isConnected()));
    }
} 